package cn.simon.free.redis.mq.core.consumer.task;

import cn.simon.free.redis.mq.annotations.RedisMessageConsumer;
import cn.simon.free.redis.mq.core.consumer.AbstractRedisMessageConsumer;
import cn.simon.free.redis.mq.enums.ConsumerType;
import cn.simon.free.redis.mq.policy.RedisMessageRejectPolicy;
import com.google.common.base.Joiner;

import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author simon
 * @date 2021/7/2 10:17
 * @copyright:
 * @modifier
 * @motto a no talent, people who have to study hard
 */
public class ThreadPoolLongPollingTask {

    /**
     * 线程池对象(单线程和多线程都用该对象,只是核心线程数不一样而已)
     * 1. 单线程核心数为1
     * 1. 多线程核心数为注解上corePoolSize()方法, {@link RedisMessageConsumer}
     */
    protected final ThreadPoolExecutor executor;
    protected final int corePollSize;
    protected final String queueName;
    protected final ConsumerType consumerType;
    protected AbstractRedisMessageConsumer abstractRedisMessageConsumerHandle;

    public ThreadPoolLongPollingTask(int corePollSize, String queueName, ConsumerType consumerType, AbstractRedisMessageConsumer abstractRedisMessageConsumerHandle) {
        this.corePollSize = corePollSize;
        this.queueName = queueName;
        this.consumerType = consumerType;
        this.abstractRedisMessageConsumerHandle = abstractRedisMessageConsumerHandle;
        this.executor = new ThreadPoolExecutor(corePollSize, corePollSize, 0, TimeUnit.MILLISECONDS, new SynchronousQueue<>(), new ThreadFactory() {
            final AtomicInteger workerThreadId = new AtomicInteger(0);

            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r, Joiner.on("").join("redis message queue[", queueName, "-", workerThreadId.getAndIncrement(), "]"));
                t.setDaemon(true);
                return t;
            }
        }, new RedisMessageRejectPolicy());
    }

    /**
     * 启动多个
     */
    public void start() {
        for (int workerThreadId = 0; workerThreadId < corePollSize; workerThreadId++) {
            this.executor.execute(LongPollingTask.builder()
                    .consumerQueue(Joiner.on("-").join(queueName, workerThreadId))
                    .consumerType(this.consumerType)
                    .longPollingConsumer(abstractRedisMessageConsumerHandle)
                    .build());
        }
    }

}
